投稿者

インターシステムズジャパン

TEAM

記事 Toshihiko Minamoto · 4月 30 8m read

AIエージェントをゼロから構築する　(1)：脳の形成

#Ensemble #InterSystems IRIS #Artificial Intelligence (AI) #Generative AI (GenAI) #ObjectScript #ツール #フレームワーク #相互運用性

ある概念は紙に書かれたままでは完璧に理解できても、他の概念は実際に手を汚すことを必要とすることがあります。例えば、運転を例に挙げましょう。エンジンの仕組みのあらゆる部品を暗記することはできますが、それが運転が実際にできることを意味するわけではありません。

実際に運転席に座り、クラッチの摩擦点や路面からの振動を身体で感じ取るまでは、その真髄を理解することはできません。コンピューティングの概念の中には直感的に理解できるものもありますが、インテリジェントエージェントは異なります。それらを理解するためには、運転席に座る必要があるのです。

これまでのAIエージェントに関する記事では、CrewAIおよびLangGraphといったツールについて取り上げてまいりました。しかし、本ガイドでは、AIエージェントのマイクロフレームワークをゼロから構築してまいります。エージェントを構築することは、単なる構文の習得を超えた取り組みであり、開発者にとって実世界の問題を解決に挑戦する貴重な旅と申せます。

とはいえ、経験そのもの以上に、これを行う根本的な理由がもう一つあります。それはリチャード・ファインマンの言葉に最もよく表れています：

「自分でつくれないものは、本当に理解しているとはいえない」

では…AIエージェントとは何でしょうか？

具体的に説明いたします。エージェントとは、本質的に目的を追求するコードです。エージェントは　「部屋の雰囲気を読み取る」能力を持ち、メールの仕分けから複雑なスケジュールの管理に至るまで、様々なタスクを実行します。

リジッドなスクリプトとは異なり、エージェントは自律性を備えています。従来のスクリプトは、現実がハードコードされたルールから外れた瞬間に機能しなくなります。しかしエージェントはそうではありません。彼らは適応します。フライトがキャンセルされても、エラーで停止することはありません。単に経路を変更するのです。

私はアーキテクチャを生物学的システムとして視覚化することを好みます：

手：ツールです。実行環境やAPIがなければ、脳は瓶に閉じ込められたままです。
神経系：オーケストレーション層です。状態を管理し、メモリを記録します。
身体：信頼性と稼働時間を保証する導入インフラストラクチャです。

単一のエージェントは確かに印象的かもしれませんが、真の力はエージェント型AIにこそ存在します。これは、複数の専門化されたエージェントが協力して共通の目標を達成する、システム全体を指します。

本質的にはデジタルエージェンシーのようなものです。調査を行うエージェントが一つ、コピーを起草するエージェントが一つ、そしてエージェント同士が互いの足を引っ張らないようにする「マネージャー」ノードが存在します。

とはいえ、正直なところ、理論だけでは限界があります。しかし、実際にこのプロジェクトを構築したいという思いが抑えきれません。さあ、手を汚しましょう。このプロジェクトをMAISと名付けました。これは二重の目的を兼ねています。技術的には「マルチエージェント相互運用システム」を意味します。しかし、ポルトガル語では単に「プラス」を意味します。これは、私たちがより多くを求め続ける姿勢への賛辞でもあります。

脳：LiteLLMによる不可知論的インテリジェンス

当社のエージェントを駆動するためには柔軟性が必要ですが、OpenAIのような特定のプロバイダーをハードコーディングすると制約が生じます。例えばGemini 3.0をテストしたい場合や、クライアントがOllama経由でLlama 3をローカルで実行することを希望される場合にはどうすればよいでしょうか。

その目的を達成するため、私はThe Musketeersプロジェクトにおいて定番となっているライブラリ、LiteLLMを利用することを好みます。

LiteLLMの優れた点は、その標準化にあります。100以上のプロバイダーにわたるリクエストとレスポンスを正常化する、ユニバーサルアダプターとしての役割を果たします。この抽象化はマルチエージェントシステムにとって極めて重要であり、エージェントの特定のニーズに基づいてモデルを自由に組み合わせて使用することを可能にします。次のようなシナリオを想像してみましょう：

第１のエージェントは、高速でコスト効率の高いモデル（例：gpt-4o-mini）を使用します。
第2のエージェントは、高度な思考能力と大きなコンテキストウィンドウが広いモデル（例：claude-3-5-sonnet）を活用し、複雑なデータの解析をいたします。

このアーキテクチャでは、設定内の文字列を変更するだけによって、エージェントが使用するモデルを定義することが可能です。

セキュリティ最優先：APIキーの取り扱い

これらのプロバイダーへの接続にはAPIキーが必要であり、ソースコードにシークレットをハードコードすることは絶対に避けたいところです。この対応における「InterSystemsの方式」は、本番認証情報を介して行う方法です。

キーの保護を確実にするため、LLMアダプターはIRISのセキュアな資格証明ストレージへのブリッジとして機能します。この引き継ぎを管理するために、APIKeysConfigというプロパティを利用します。 LiteLLMで必要とされるプロバイダー固有のキー名（例：OPENAI_API_KEY、AZURE_API_KEY）を、カンマで区切ってこのプロパティに入力する必要があります。

アダプターが初期化されると、セキュアストレージから実際のシークレットを取得し、環境変数として割り当てます。これにより、LiteLLMはコード内で生の鍵を公開することなく認証を行うことが可能となります：

Method OnInit() As %Status
{
    Set tSC = $$$OK
    Try {
        Do ..Initialize()
    } Catch e {
        Set tSC = e.AsStatus()
    }
    Quit tSC
}

/// Python環境におけるAPIキーの設定
Method Initialize() [ Language = python ]
{
    import os
    for tKeyName in self.APIKeysConfig.split(','): 
        credential = iris.cls("Ens.Config.Credentials")._OpenId(tKeyName.strip())
        if not credential:
            continue
        os.environ[tKeyName] = credential.PasswordGet()
}

セキュリティ層が整いましたので、次にアダプターの核心的な思考プロセスに焦点を当てましょう。ここでは、どのモデルを呼び出すか、またメッセージをどのように構成するかを定義します。アダプターには、デフォルトのモデルを決定する設定プロパティが設定されています：

/// リクエストで指定されていない場合には、デフォルトモデルとして使用する
Property DefaultModel As %String [ InitialExpression = "gpt-4o-mini" ];

ただし、魔法は実行時に起こります。入力メッセージ dc.mais.messages.LLMRequest には、オプションの Model プロパティがあります。オーケストレーター（BPL）がこのプロパティを指定して送信した場合、アダプターはその動的な選択を尊重します。そうでない場合は、DefaultModel にフォールバックします。

入力と指示の分離

もう一つの重要な設計決定は、LLMへのテキスト送信方法です。生の文字列をそのまま送信する代わりに、リクエスト内で概念を2つのフィールドに分割しました：

Content： ここに「システムプロンプト」または現在のエージェントの指示を入力します（例：「あなたはワインの専門家であるウェイターです…」）。
UserContent： ここにユーザーの実際の入力内容を入力します（例：「魚料理に合うワインは何ですか？」）。

これにより、LiteLLM向けに明確なメッセージ配列を構築することが可能となり、AIが自身のペルソナとユーザーの質問を明確に区別できるようになります。

以下に、主要なCallLiteLLMメソッドが、IRIS内で直接Pythonを使用してこのパズルを組み立てる方法をご説明いたします：


Method CallLiteLLM(pRequest As dc.mais.messages.LLMRequest) As dc.mais.messages.LLMResponse [ Language = python ]
{
    import litellm
    import json
    import time
    import iris

    t_attempt = 0
    max_retries = self.MaxRetries
    retry_delay = self.RetryDelay

    last_error = None
    pResponse = iris.cls("dc.mais.messages.LLMResponse")._New()

    while t_attempt <= max_retries:
        t_attempt += 1

        try:
            model = pRequest.Model
            if not model:
                model = self.GetDefaultModel()

            messages = [{"role": "user", "content": pRequest.Content}]
            
            if pRequest.UserContent:
                messages.append({"role": "user", "content": pRequest.UserContent})

            response = litellm.completion(model=model, messages=messages)

            pResponse.Model = response.model

            choices_list = []
            if hasattr(response, 'choices'):
                for choice in response.choices:
                    if hasattr(choice, 'model_dump'):
                        choices_list.append(choice.model_dump())
                    elif hasattr(choice, 'dict'):
                        choices_list.append(choice.dict())
                    else:
                        choices_list.append(dict(choice))
            
            pResponse.Choices = json.dumps(choices_list)
            if (len(response.choices) > 0 ):
                pResponse.Content = response.choices[0].message.content

            if hasattr(response, 'usage'):
                if hasattr(response.usage, 'model_dump'):
                    pResponse.Usage = json.dumps(response.usage.model_dump())
                else:
                    pResponse.Usage = json.dumps(dict(response.usage))

            if hasattr(response, 'error') and response.error:
                pResponse.Error = json.dumps(dict(response.error))

            return pResponse

        except Exception as e:
            last_error = str(e)
            
            class_name = "dc.mais.adapter.LiteLLM"
            iris.cls("Ens.Util.Log").LogError(class_name, "CallLiteLLM", f"LiteLLM call attempt {t_attempt} failed: {last_error}")

            if t_attempt > max_retries:
                break

            time.sleep(retry_delay)

    error_payload = {
        "message": "All LiteLLM call attempts failed",
        "details": last_error
    }
    pResponse.Error = json.dumps(error_payload)

    return pResponse
}

クラシックな構文を好まれる方のために、同じメソッドのObjectScript版も用意しました：

Method CallLiteLLMObjectScript(pRequest As dc.mais.messages.LLMRequest, Output pResponse As dc.mais.messages.LLMResponse) As %Status
{
    Set tSC = $$$OK
    Set tAttempt = 0
    Set pResponse = ##class(dc.mais.messages.LLMResponse).%New()
    
    While tAttempt <= ..MaxRetries {
        Set tAttempt = tAttempt + 1
        
        Try {
            Set model = $Select(pRequest.Model '= "": pRequest.Model, 1: ..GetDefaultModel())
            
            // 歴史の準備
            Set jsonHistory = [].%FromJSON(pRequest.History)
            Set:(jsonHistory="") jsonHistory = []
            
            // ツール出力がある場合はそれを注射する（ループを閉じるロジック）
            If (pRequest.ToolCallId '= "") && (pRequest.ToolOutput '= "") {
                Set tToolMsg = {}
                Set tToolMsg.role = "tool"
                Set tToolMsg.content = pRequest.ToolOutput
                Set tToolMsg."tool_call_id" = pRequest.ToolCallId
                Do jsonHistory.%Push(tToolMsg)
            }
            
            // 現在のユーザーコンテンツ／プロンプトが空でない場合は追加する
            If (pRequest.Content '= "") {
                Do jsonHistory.%Push({"role": "user", "content": (pRequest.Content)})
            }
            
            // 追加のユーザーコンテンツフィールドが存在する場合は追加する
            If (pRequest.UserContent'="") {
                Do jsonHistory.%Push({"role": "user", "content": (pRequest.UserContent)})
            }
            
            Set strMessages = jsonHistory.%ToJSON()

            // Pythonヘルパーを呼び出す
            Set tResponse = ..PyCompletion(model, strMessages, pRequest.Parameter, 1)
        
            // マッピング応答
            Set pResponse.Model = tResponse.model
            
            // Pythonの選択をIRIS DynamicArrayに変換する
            Set choices = []
            For i=0:1:tResponse.choices."__len__"()-1 {
                Do choices.%Push({}.%FromJSON(tResponse.choices."__getitem__"(i)."to_json"()))
            }
            Set pResponse.Choices = choices.%ToJSON()
            
            // 最後の選択をプロセスする
            If (choices.%Size()>0) {
                Set choice = choices.%Get(choices.%Size()-1)
                
                If ($IsObject(choice.message)){
                    Set pResponse.Content = choice.message.content
                    
                    // ツールコール抽出する
                    // 「tool_calls」が存在し、有効なオブジェクト（DynamicArray）であるかどうかを確認する
                    Set tToolCalls = choice.message."tool_calls"
                    
                    // それがオブジェクト（配列）であり、空文字列ではないことを確認する
                    If $IsObject(tToolCalls) {
                        Do ..GetToolCalls(tToolCalls, .pResponse)
                    }
                }
            }

            // マッピングの使用方法
            If ..hasattr(tResponse, "usage") {
                Set pResponse.Usage = {}.%FromJSON(tResponse.usage."to_json"()).%ToJSON()
            }
            
            // サクセス - ループ終了
            Quit 

        } Catch e {
            Set tSC = e.AsStatus()
            $$$LOGERROR("LiteLLM call attempt "_tAttempt_" failed: "_$System.Status.GetOneErrorText(tSC))
            If tAttempt > ..MaxRetries Quit
            Hang ..RetryDelay
        }
    }
    
    If ($$$ISERR(tSC)) {
        Set pResponse.Error = {
            "message": "All LiteLLM call attempts failed",
            "details": ($System.Status.GetOneErrorText(tSC))
        }.%ToJSON()
    }
    Quit tSC
}

このロジックのObjectScript版において、..PyCompletion(...)への呼び出しが記述されていることに、お気づきかもしれません。これは標準のシステムメソッドではなく、二つの言語間のデータマーシャリングを処理するために作られたカスタムヘルパーです。 IRISではPythonへの直接呼び出しが可能ですが、複雑な入れ子構造（特定のデータ型を含むオブジェクトのリストなど）を渡す際には、手動での変換が必要となる場合があります。 PyCompletionメソッドは、変換層として機能します。ObjectScriptからのデータをシリアライズされたJSON文字列として受け取ります。その後、Python環境内でそれらをデシリアライズし、ネイティブのPython辞書およびリストに変換します（json.loadsを使用）。最後に、LiteLLMリクエストを実行します。この「ハイブリッド」アプローチにより、ObjectScriptコードはビジネスロジック（ループ処理、履歴管理）に完全に集中し、クリーンで読みやすい状態を保ちます。一方で、データ型の変換やライブラリとの相互作用といった負荷の高い処理は、専用の小さなPythonラッパーに委ねられます。

このシンプルな構造により、強力な制御が可能となります。BPLがフローの各ステップで動的に操作の中核（モデル）や個性を担う部分（コンテンツ）を切り替えられる一方で、アダプターが技術的な「配管」の役割を担います。

舞台は整っておりますが、空っぽでございます。

これまで多くの課題を解決してまいりました。PythonとLiteLLMを用いて、LLMへの安全でプロバイダー非依存の接続基盤を構築いたしました。**kwargsを用いた複雑な相互運用性の課題を解決し、IRISネイティブストレージの活用により、資格情報を安全に扱う方法を確立いたしました。

しかしながら、よくご覧になればお分かりいただけるように、私たちは美しい車と強力なエンジンを備えておりますが、運転手がおりません。

「脳」へのリンクは確立いたしましたが、明確なペルソナが欠如しております。GPTを起動することは可能ですが、具体的な指示がない限り、役立つグリーターとして振る舞うべきか、技術サポートエンジニアとして対応すべきかを判断できません。現状では、メモリを保持しないステートレスなプロセッサに過ぎず、目標を欠き、いかなるツールとも接続されていない状態です。

さて、第2部では、この脳に魂を吹き込みます。具体的には以下のことを行います：

ペルソナを定義するため、dc.mais.adapter.Agent クラスを構築します。
ビジネスルールを強制するため、ダイナミックプロンプトエンジニアリングを習得します。
エージェント間通信のための**「許可リスト」**ロジックを実装します。
エージェントを真にスマートにするReActパラダイム理論を深く学びます。

アダプターを複雑にしすぎてしまったでしょうか？環境変数の処理について、より簡潔な方法はお持ちでしょうか？もしそうでしたら、あるいは第2部に進む前に私のロジックに誤りをお気づきになりましたら、ぜひ下記のコメントで指摘してくださいますようお願いいたします。この記事は、皆様と知識を共有するのと同様に、私自身も皆様から学ぶために執筆しております。

謝辞： Musketeerの仲間である@José Pereira様に、LiteLLMの素晴らしさを紹介していただきましたことに、心より感謝申し上げます。

ご期待ください。これからが本番です。

InterSystems Open Exchangeで関連アプリケーションを確認してください

元の記事へさんが書いた @Henry Pereira

ディスカッション (0)0

続けるにはログインするか新規登録を行ってください

返信を追加する